package preprocess;

import io.TextInputFormatForSequenceFile;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import trouger.dmt.JoinTableRunner;
import trouger.dmt.JoinType;
import utils.Utils;

public class ClickLogItemIdConvertor implements Tool{
	
	public static String getClickLogPath(String date) {
		//return "D:\\dev\\clickprediction\\data\\ClickLog";
		return Utils.rootDir + date + "/ClickLog2/";
	}
	
	@Override 
	public Configuration getConf() {
		return null;
	}
	
	@Override
	public void setConf(Configuration conf) {
		
	}
	/*
	 * (non-Javadoc)
	 * UserFiltered Table
	 * Path: /group/tbsc-dev/xzwang/clickprediction/$date/ClickLog2/
	 * 1.user 32位id
	 * 2.Query 
	 * 3.bucket号（占用Query_type，如果对应不到bucket号，则为空，其它为0-19的值）
	 * 4.User 数字id
	 * 5.用户分类
	 * 6.Machine id
	 * 7.Session ID
	 * 8.点击时间
	 * 9.搜索的来源pid
	 * 10.搜索类目
	 * 11.上一步点击模块ID（智能导航或相关搜索）
	 * 12.排序, 格式以转换
	 * 13.列表模式（大图、小图）
	 * 14.消保筛选
	 * 15.所在地筛选
	 * 16.每页商品数
	 * 17.第几页
	 * 18.被点击的商品在商品列表的第几个位置
	 * 19.点击商品的auctionId(数字)
	 * 20.商品所属类目，用auctionId关联
	 * 21.商品所属属性，用auctionId关联 (目前Hadoop上没有该字段，需要导入)
	 * 22.商品所属的spuid
	 * 23.商品所属卖家ID
	 * 24.B/C商品标识
	 * 25.搜索属性
	 * 26.Tab类型 (0:未知，1：所有宝贝，2：人气，3：淘宝商城，4：二手)
	 * 27.是否豆腐块商品（0:否，1：是） 
	 * 注：wiki中描述有28列是错的，实际只有27列，最后一列无效
	 * */	 
	@Override
	public int run(String[] args) throws Exception {		
		String clickLog = DumpClickLog.getClickLogPath(args[0]);
		String auctionFile = "/group/taobao/taobao/hive/r_auction_auctions/pt=" + 
			args[0] + "000000/if_online=1";
		String output = getClickLogPath(args[0]);
		System.out.println("start job ClickLogItemIdConvertor...");
		System.out.println("auctionFile: " + auctionFile);
		System.out.println("clickLog: " + clickLog);
		System.out.println("output: " + output);
		if (Utils.exitSystem(args))
			return -1;
		JoinTableRunner r = new JoinTableRunner();
		r.addSourceTable(auctionFile, TextInputFormatForSequenceFile.class, "UTF-8", "\u0001");
		r.addSourceTable(clickLog, "UTF-8", "\t");
		r.addKeyGroup("1.1, 2.19");
		r.setJoinType(JoinType.InnerJoin);	
		StringBuilder col = new StringBuilder();
		for (int i = 1; i <= 18; ++i)
			col.append("2." + String.valueOf(i) + ", ");
		col.append("1.19, ");
		for (int i = 20; i < 27; ++i)
			col.append("2." + String.valueOf(i) + ", ");
		col.append("2.27");
		r.setResultFieldsEx(col.toString());
		r.setResultTable(output);
		r.run("ClickLogItemIdConvertorJob");
		return 0;		 	
	}
	
	/**
	 * @param args
	 */
	public static void main(String[] args) throws Exception {
		Utils.checkArgs(args);
		
		// TODO Auto-generated method stub
		int res = ToolRunner.run(new ClickLogItemIdConvertor(), args);
		System.exit(res);
	}
}
